package com.huangyi.rabbitmq_amqp.publish;

import com.huangyi.rabbitmq_amqp.config.Constants;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmListener;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

import java.io.IOException;
import java.util.Collections;
import java.util.SortedSet;
import java.util.TreeSet;
import java.util.concurrent.TimeoutException;

public class Producer {
    public static  Integer MESSAGE_COUNT = 100000;

    public static void main(String[] args) throws IOException, InterruptedException, TimeoutException {
//        //单独确认机制
//        publishingMessagesIndividually();
        //批量确认
        publishingMessagesInBatches();
        //异步确认
        handlingPublisherConfirmsAsynchronously();
    }

    private static void handlingPublisherConfirmsAsynchronously() throws IOException, TimeoutException, InterruptedException {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost(Constants.HOST);
        factory.setPort(Constants.PORT);
        factory.setVirtualHost(Constants.VIRTUAL_HOST);
        factory.setUsername(Constants.USER);
        factory.setPassword(Constants.PASSWORD);

        Connection connection = factory.newConnection();
        Channel channel  = connection.createChannel();
        channel.confirmSelect();
        channel.queueDeclare(Constants.PUBLISHER_CONFIRMS_QUEUE3,true,false,false,null);
        long startTime = System.currentTimeMillis();
        SortedSet<Long> confirmSeqNo = Collections.synchronizedSortedSet(new TreeSet<>());
        channel.addConfirmListener(new ConfirmListener() {
            @Override
            public void handleAck(long l, boolean b) throws IOException {
                if(b){
                    confirmSeqNo.headSet(l+1).clear();
                }else{
                    confirmSeqNo.remove(l);
                }
            }

            @Override
            public void handleNack(long l, boolean b) throws IOException {
                handleAck(l,b);
                System.out.println("丢失数据" + l);
            }
        });
        for (int i = 0; i < MESSAGE_COUNT; i++) {
            String msg = "hello publisher confirms"+i;
            long seqNo = channel.getNextPublishSeqNo();
            channel.basicPublish("",Constants.PUBLISHER_CONFIRMS_QUEUE3, null, msg.getBytes());
            confirmSeqNo.add(seqNo);
        }
        while (!confirmSeqNo.isEmpty()){
            Thread.sleep(10);
        }
        long endTime = System.currentTimeMillis();
        System.out.println("异步确认最终耗时" + (endTime - startTime) + "ms");

    }

    private static void publishingMessagesInBatches() throws IOException, TimeoutException, InterruptedException {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost(Constants.HOST);
        factory.setPort(Constants.PORT);
        factory.setVirtualHost(Constants.VIRTUAL_HOST);
        factory.setUsername(Constants.USER);
        factory.setPassword(Constants.PASSWORD);

        Connection connection = factory.newConnection();
        Channel channel  = connection.createChannel();
        channel.confirmSelect();
        channel.queueDeclare(Constants.PUBLISHER_CONFIRMS_QUEUE2,true,false,false,null);
        long startTime = System.currentTimeMillis();
        Integer count  = 0;
        for (int i = 0; i <MESSAGE_COUNT ; i++) {
            count++;
            String msg = "Massage" + i;
            channel.basicPublish("", Constants.PUBLISHER_CONFIRMS_QUEUE2, null, msg.getBytes());
            if(count == 200){
                channel.waitForConfirmsOrDie(5000);
                count = 0;
            }
        }
        if (count > 0) {
            channel.waitForConfirmsOrDie(500);
        }
        long endTime = System.currentTimeMillis();
        System.out.println("批量确认最终耗费时间" + (endTime - startTime) + "ms");

    }


    private static void publishingMessagesIndividually() throws IOException, TimeoutException, InterruptedException {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost(Constants.HOST);
        factory.setPort(Constants.PORT);
        factory.setVirtualHost(Constants.VIRTUAL_HOST);
        factory.setUsername(Constants.USER);
        factory.setPassword(Constants.PASSWORD);

        Connection connection = factory.newConnection();
        Channel channel  = connection.createChannel();
        channel.confirmSelect();
        channel.queueDeclare(Constants.PUBLISHER_CONFIRMS_QUEUE1,true,false,false,null);
        long startTime = System.currentTimeMillis();
        for (int i = 0; i < 100; i++) {

            String msg = "Message " + i;
            channel.basicPublish("",Constants.PUBLISHER_CONFIRMS_QUEUE1,null,msg.getBytes());
            channel.waitForConfirmsOrDie(5000);

        }
        long endTime = System.currentTimeMillis();
        System.out.println("单独确认最终耗费时间：" + (endTime - startTime) + "ms");
    }

}
